package rabbitmq

import (
	"fmt"
	"news_for_deepseek_interface/config"
	"sync"
	"time"

	"github.com/streadway/amqp"
)

var (
	rabbitmqClient *amqp.Connection   // RabbitMQ连接实例
	channels       chan *amqp.Channel // RabbitMQ通道池
	initOnce       sync.Once          // 初始化锁
	closeOnce      sync.Once          // 关闭锁
)

const (
	channelPoolSize = 500  // 预创建的通道数量
	maxChannelSize  = 2000 // 最大通道数量
)

func InitRabbitMQ() error {
	var err error

	// 使用 initOnce 确保只初始化一次
	initOnce.Do(func() {
		// 连接RabbitMQ
		url := fmt.Sprintf("amqp://%s:%s@%s:%d/%s",
			config.DefaultConfig.Rabbitmq.User,
			config.DefaultConfig.Rabbitmq.Password,
			config.DefaultConfig.Rabbitmq.Host,
			config.DefaultConfig.Rabbitmq.Port,
			config.DefaultConfig.Rabbitmq.Vhost,
		)
		rabbitmqClient, err = amqp.Dial(url)
		if err != nil {
			err = fmt.Errorf("连接RabbitMQ失败: %v", err)
			return
		}

		// 初始化通道池
		channels = make(chan *amqp.Channel, channelPoolSize)

		// 预创建一部分通道
		for i := 0; i < channelPoolSize; i++ {
			ch, e := createChannel()
			if e != nil {
				err = fmt.Errorf("预创建通道失败: %v", e)
				return
			}
			channels <- ch
		}

		// 获取一个通道用于声明交换机和队列
		ch, e := GetChannel()
		if e != nil {
			err = fmt.Errorf("获取通道失败: %v", e)
			return
		}
		defer ReleaseChannel(ch)

		// 声明交换机和队列并绑定
		for _, queue := range queues {
			if e := DeclareExchangeAndQueue(ch, queue); e != nil {
				err = fmt.Errorf("声明交换机和队列失败: %v", e)
				return
			}
		}
	})

	return err
}

// DeclareExchangeAndQueue 声明交换机和队列并绑定
func DeclareExchangeAndQueue(ch *amqp.Channel, config QueueConfig) error {
	// 声明交换机
	if err := ch.ExchangeDeclare(
		config.ExchangeName,
		config.Kind,
		true,
		false,
		false,
		false,
		nil,
	); err != nil {
		return fmt.Errorf("声明交换机 %s 失败: %v", config.ExchangeName, err)
	}

	// 声明队列
	if _, err := ch.QueueDeclare(
		config.QueueName,
		true,
		false,
		false,
		false,
		nil,
	); err != nil {
		return fmt.Errorf("声明队列 %s 失败: %v", config.QueueName, err)
	}

	// 绑定队列
	if err := ch.QueueBind(
		config.QueueName,
		config.RoutingKey,
		config.ExchangeName,
		false,
		nil,
	); err != nil {
		return fmt.Errorf("绑定队列 %s 失败: %v", config.QueueName, err)
	}

	return nil
}

// createChannel 创建通道
// @return *amqp.Channel 通道实例
// @return error 创建失败错误信息
func createChannel() (*amqp.Channel, error) {
	ch, err := rabbitmqClient.Channel()
	if err != nil {
		return nil, fmt.Errorf("创建通道失败: %v", err)
	}

	// 设置QoS 预取数量
	err = ch.Qos(50, 0, false)
	if err != nil {
		ch.Close()
		return nil, fmt.Errorf("设置QoS失败: %v", err)
	}

	return ch, nil
}

func GetChannel() (*amqp.Channel, error) {
	if channels == nil {
		return nil, fmt.Errorf("RabbitMQ未初始化")
	}

	select {
	case ch := <-channels:
		// 检查通道是否可用
		if err := ch.Flow(true); err != nil {
			// 如果通道不可用，则创建新的通道
			return createChannel()
		}
		return ch, nil
	default:
		// 池空时快速创建新通道
		return createChannel()
	}
}

// ReleaseChannel 释放通道
// @param ch *amqp.Channel 通道实例
func ReleaseChannel(ch *amqp.Channel) {
	if ch != nil {
		select {
		case channels <- ch:
			// 将通道放回池中
		default:
			// 通道池已满，直接关闭通道
			ch.Close()
		}
	}
}

// Publish 发送消息到指定的交换机和路由
func Publish(exchange, routingKey string, body []byte) error {
	// 获取通道
	ch, err := GetChannel()
	if err != nil {
		return fmt.Errorf("获取通道失败: %v", err)
	}
	defer ReleaseChannel(ch)

	// 发送消息
	err = ch.Publish(
		exchange,   // 交换机名称
		routingKey, // 路由键
		false,      // mandatory
		false,      // immediate
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         body,
			DeliveryMode: amqp.Persistent, // 消息持久化
			Timestamp:    time.Now(),
		},
	)

	if err != nil {
		return fmt.Errorf("发送消息失败: %v", err)
	}

	return nil
}
